package spring.cloud.tasks.client.item_task_processor;

import com.alibaba.fastjson.JSONObject;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import spring.cloud.tasks.client.ItemTaskWorker;
import spring.cloud.tasks.client.item_task_processor.task_item.ClientTaskItemContractExtensionInfo;
import spring.cloud.tasks.client.util.CronExpression;
import spring.cloud.tasks.common.utils.ApplicationContextUtils;
import spring.cloud.tasks.common.utils.Uuids;
import spring.cloud.tasks.common.zookeeper.ZooKeeperDataManager;
import spring.cloud.tasks.common.zookeeper.ZooKeeperItemTaskDataManager;
import spring.cloud.tasks.common.zookeeper.ZookeeperNodeConstants;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Date;
import java.util.List;
import java.util.Timer;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j

public class TaskProcessor {
    @Setter
    @Getter
    protected volatile Task task;

    @Setter
    @Getter
    protected volatile TaskProcessorInfo taskProcessorInfo;
    @Setter
    @Getter
    protected volatile String threadId;
    @Setter
    @Getter
    private volatile int threadCountPerTaskProcessor = 5;
    @Setter
    @Getter
    protected volatile String threadRuntimeErrorInfo = null;//调度器主线程运行时的异常
    @Setter
    @Getter
    public volatile ZooKeeperItemTaskDataManager zooKeeperItemTaskDataManager;
    @Setter
    @Getter
    volatile ItemTaskWorker itemTaskWorker;
    @Setter
    @Getter
    public volatile boolean needPongPong = true;
    @Setter
    @Getter
    private volatile Timer heartBeatTimer;
    @Setter
    @Getter
    private volatile Date nextRunTime;
    @Setter
    @Getter
    protected volatile Lock processorLock = new ReentrantLock();
    //
    @Setter
    @Getter
    volatile TaskProcessingThreadContext taskProcessingThreadContext;
    @Setter
    @Getter
    volatile spring.cloud.tasks.client.item_task_processor.api.TaskProcessor taskProcessor;
    volatile Statistics statistics = new Statistics();
    @Setter
    @Getter
    protected volatile long lastprocessorsStatVersionInZooKeeper = -1;
    @Setter
    @Getter
    protected volatile int taskItemListSizeInZooKeeper = 0;//所有的taskItemCount
    @Setter
    @Getter
    protected volatile boolean needReloadTaskItemIdAndParameterList = true;
    @Setter
    @Getter
    private final Object needReloadTaskItemIdAndParameterListLock = new Object();
    //@Setter
    //@Getter 存在同名的方法
    protected volatile List<String> taskItemList = new CopyOnWriteArrayList<String>();

    //
    @Setter
    @Getter
    protected volatile boolean stop = false;
    @Setter
    @Getter
    volatile boolean pause = true;
    @Setter
    @Getter
    volatile String pauseMessage = "";
    @Setter
    @Getter
    volatile String runOrRecoverMessage = "";
    @Setter
    @Getter
    protected volatile boolean exceptionOccurred = false;//发生异常<自标记>

    //在休眠模式下,任务执行过程中评估来当前任务执行不及休眠下,会和下次任务调度时间可能很近则不需要销毁自己的资源,一直执行下去,直到任务评估可以进行销毁的时候。
    public volatile boolean needReInitProcessor = true;
    @Setter
    @Getter
    protected volatile ConcurrentHashMap<String, ClientTaskItemContractExtensionInfo> taskItemToClientTaskItemContractExtensionInfoMap = new ConcurrentHashMap<String, ClientTaskItemContractExtensionInfo>();

    @Setter
    @Getter
    protected volatile ReentrantLock lock = new ReentrantLock();

    public TaskProcessor(String taskId, ItemTaskWorker itemTaskWorker, ZooKeeperDataManager zooKeeperItemTaskDataManager, int threadCountPerTaskProcessor) throws Exception {
        log.info("create ItemTaskProcessor for task:" + taskId);
        this.task = this.zooKeeperItemTaskDataManager.loadTask(taskId);
        Object bean = ApplicationContextUtils.getApplicationContext().getBean(TaskIdToBeanIdFactory.getBeanId(taskId));
        this.taskProcessor = (spring.cloud.tasks.client.item_task_processor.api.TaskProcessor) bean;
        //
        this.itemTaskWorker = itemTaskWorker;
        this.zooKeeperItemTaskDataManager = (ZooKeeperItemTaskDataManager) zooKeeperItemTaskDataManager;
        this.threadCountPerTaskProcessor = threadCountPerTaskProcessor;
        this.taskProcessorInfo = createItemTaskProcessorInfo(taskId, itemTaskWorker, this.zooKeeperItemTaskDataManager, threadCountPerTaskProcessor);


        this.zooKeeperItemTaskDataManager.registerProcessor(this.taskProcessorInfo);//?为什么关联
        this.heartBeatTimer = new Timer(TaskProcessorHeartBeatTimerTask.class.getName() + "-" + this.taskProcessorInfo.getTaskId() + "-" + this.taskProcessorInfo.getProcessorId());
        this.heartBeatTimer.schedule(new TaskProcessorHeartBeatTimerTask(this), new java.util.Date(System.currentTimeMillis() + 500), this.task.getHeartBeatRate());
        newThread();
    }


    public static TaskProcessorInfo createItemTaskProcessorInfo(String taskId, ItemTaskWorker itemTaskWorker, ZooKeeperItemTaskDataManager zooKeeperItemTaskDataManager, int threadNum)//
    {
        TaskProcessorInfo taskProcessorInfo = new TaskProcessorInfo();
        try {
            taskProcessorInfo.ip = InetAddress.getLocalHost().getHostAddress();
        } catch (UnknownHostException e) {
            e.printStackTrace();
        }
        taskProcessorInfo.taskId = taskId;
        taskProcessorInfo.workerId = itemTaskWorker.getWorkerId();
        taskProcessorInfo.processorId = taskId + ZookeeperNodeConstants.$ + itemTaskWorker.getWorkerId() + ZookeeperNodeConstants.$ + (Uuids.getUuid());//主要是uuid可能有问题
        taskProcessorInfo.threadNum = threadNum;
        taskProcessorInfo.registerTime = new Date(zooKeeperItemTaskDataManager.zooKeeperCurrentTimeMillis());
        taskProcessorInfo.heartBeatTime = null;
        taskProcessorInfo.mark = "";
        taskProcessorInfo.version = 0;
        return taskProcessorInfo;
    }

    public void newThread() throws Exception {
        threadId = TaskProcessor.class.getName() + "-" + this.taskProcessorInfo.processorId;
        Thread thread = new Thread(threadId) {

            public void run() {
                try {
                    log.info(taskProcessorInfo.getProcessorId() + "开始启动");


                    while (true) {
                        if (stop == true) {
                            log.warn("停止调度器:" + taskProcessorInfo.getProcessorId());
                            return;
                        }
                        refreshProcessorInfo();
                        //再次被进行重新初始化
                        boolean taskItemsIsInited = zooKeeperItemTaskDataManager.taskItemsIsInited(task.getTaskId());
                        if (!taskItemsIsInited) {
                            TimeUnit.SECONDS.sleep(10);//存在分配task item ，需要等待stop被设置为true.
                            continue;
                        }
                        if (getTaskItemList().size() > 0) {
                            taskItemListSizeInZooKeeper = zooKeeperItemTaskDataManager.loadTaskItemList(taskProcessorInfo.getTaskId()).size();
                            break;
                        }
                        TimeUnit.SECONDS.sleep(1);
                    }
                    log.info("调度器:" + taskProcessorInfo.getProcessorId() + "拉取任务:" + JSONObject.toJSONString(taskItemList));
                    startDealWithTasks();
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                    threadRuntimeErrorInfo = e.getMessage();
                    exceptionOccurred = true;//出现异常
                }
            }//异步
        };
        thread.setDaemon(true);
        thread.start();
    }


    public void refreshProcessorInfo() throws Exception {
        processorLock.lock();
        try {
            if (this.stop == true) {
                return;
            }
            StringBuilder sb = new StringBuilder();
            sb.append("暂停信息:");
            if (pauseMessage != null) {
                sb.append(pauseMessage);
            } else {
                sb.append("-");
            }
            sb.append(";");
            sb.append("线程运行时异常:");
            if (threadRuntimeErrorInfo != null) {
                sb.append(threadRuntimeErrorInfo);
            } else {
                sb.append("-");
            }
            sb.append(";");
            this.taskProcessorInfo.setMark(sb.toString());
            //
            if (!this.zooKeeperItemTaskDataManager.updateProcessorHeartBeatTimeAndVersion(this.taskProcessorInfo)) {//可能异常
                this.clearTaskItemAndClearTaskDataIdList();
            }
        } catch (Exception e) {
            exceptionOccurred = true;
        } finally {
            processorLock.unlock();
        }
    }

    public void unRegisterProcessor() throws Exception {
        processorLock.lock();
        try {
            this.zooKeeperItemTaskDataManager.unRegisterProcessor(this.taskProcessorInfo);
        } finally {
            processorLock.unlock();
        }
    }

    /**
     * 重启,该功能可以进行任务处理器初始化
     */
    public void run(String message) throws Exception {
        if (this.pause == true) {
            //
            refreshProcessorInfo();
            //
            this.pause = false;
            //
            this.runOrRecoverMessage = message;
            //
            this.taskProcessingThreadContext = new TaskProcessingThreadContext(this, taskProcessor, this.statistics);

        }
    }

    /**
     * 暂停任务执行
     */
    public void pause(String message) throws Exception {
        if (!this.pause) {
            this.pause = true;
            this.pauseMessage = message;
            //
            if (this.taskProcessingThreadContext != null) {
                this.taskProcessingThreadContext.destroy();
                this.taskProcessingThreadContext = null;
            }
        }
    }

    public void stop() throws Exception {
        needPongPong = false;//不再进行心跳
        //
        this.pause = true;//用于控制调度器是否暂停
        this.stop = true;//用于控制调度器是否停止
        //
        if (this.taskProcessingThreadContext != null) {
            this.taskProcessingThreadContext.destroy();
            this.taskProcessingThreadContext = null;
        }
        //
        this.unRegisterProcessor();
        this.heartBeatTimer.cancel();

    }


    /**
     * 任务执行完毕后才进行拉取数据
     */
    public List<String> getTaskItemAndParameterListUntilTaskDataListIsAllFinshed() {
        try {
            if (this.needReloadTaskItemIdAndParameterList) {
                if (this.taskProcessingThreadContext != null) {
                    while (!this.taskProcessingThreadContext.noTaskDatas()) {
                        TimeUnit.SECONDS.sleep(1);
                    }//must all task is all finshed
                }
                synchronized (needReloadTaskItemIdAndParameterListLock) {
                    this.getTaskItemList();
                    this.needReloadTaskItemIdAndParameterList = false;
                }
            }

            return this.taskItemList;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 拉取processor对应的task item and parameter。
     */
    protected synchronized List<String> getTaskItemList() throws Exception {
        this.lastprocessorsStatVersionInZooKeeper = this.zooKeeperItemTaskDataManager.getProcessorsStatVersion(this.taskProcessorInfo.getTaskId());
        try {
            //调度器释放TaskItem处理
            this.zooKeeperItemTaskDataManager.releaseContractExtensionInfos(this.taskProcessorInfo, taskItemToClientTaskItemContractExtensionInfoMap);
            this.taskItemList.clear();
            //
            this.taskItemList = this.zooKeeperItemTaskDataManager.getTaskItemAndParameterList(this.taskProcessorInfo);
            return this.taskItemList;
        } catch (Throwable e) {
            this.lastprocessorsStatVersionInZooKeeper = -1;
            if (e instanceof Exception) {
                throw (Exception) e;
            } else {
                throw new Exception(e);
            }
        }
    }

    public boolean isContinueWhenNoTaskDataIdList() throws Exception {
        if (this.task.getCronExpressionString() == null) {
            this.needReInitProcessor = false;
            return true;//如果没有外部命令永远没有清理的机会
        } else {
            if (this.taskItemList.size() > 0) {
                this.needReInitProcessor = false;
                return true;
            } else {
                //进行比较现在的时间和下次运行的时间是否很近
                boolean close = (nextRunTime.getTime() - (new Date()).getTime()) <= 1000 * 60;
                if (close) {
                    this.needReInitProcessor = false;
                    return true;
                } else {
                    //this.taskItemList.size()==0
                    this.needReInitProcessor = true;
                    this.pause("");
                    return false;
                }

            }
        }

    }

    public void clearTaskItemAndClearTaskDataIdList() {
        try {
            this.taskItemList.clear();
            if (this.taskProcessingThreadContext != null) {
                this.taskProcessingThreadContext.clearTaskDataIdList();
            }
        } finally {
            this.needReloadTaskItemIdAndParameterList = true;
        }

    }


    //

    /**
     * 开始计划处理或者马上处理任务
     */
    public void startDealWithTasks() throws Exception {
        boolean startRunMode = false;
        if (this.task.getCronExpressionString() == null) {
            startRunMode = true;
        } else {
            String cronExpressionString = this.task.getCronExpressionString();
            if (cronExpressionString.toLowerCase().startsWith("rightNowRun:")) {
                cronExpressionString = cronExpressionString.substring("rightNowRun:".length());
                this.task.setCronExpressionString(cronExpressionString);
                startRunMode = true;
            }
            CronExpression cronExpression = new CronExpression(cronExpressionString);
            Date nowDate = new Date(this.zooKeeperItemTaskDataManager.zooKeeperCurrentTimeMillis());
            Date nextRunTime = cronExpression.getNextTimeAfter(nowDate);
            this.heartBeatTimer.schedule(new TaskProcessorScheduleTimerTask(this, this.heartBeatTimer, cronExpressionString), nextRunTime);
            this.nextRunTime = nextRunTime;
            this.taskProcessorInfo.setNextRunTime(nextRunTime);
        }
        if (startRunMode == true) {
            this.run("");
        }
    }


    public void pongpong() throws Exception {
        try {
            refreshProcessorInfo();
            //本地状态和zk状态不一致则直接拉取数据
            boolean needGetTaskItemIdAndParameterList = this.lastprocessorsStatVersionInZooKeeper < this.zooKeeperItemTaskDataManager.getProcessorsStatVersion(this.taskProcessorInfo.getTaskId());
            if (needGetTaskItemIdAndParameterList != this.needReloadTaskItemIdAndParameterList) {
                synchronized (needReloadTaskItemIdAndParameterListLock) {
                    this.needReloadTaskItemIdAndParameterList = true;
                }
            }
            //主动更新task item and parameter list
            if (this.pause || (this.taskProcessingThreadContext != null && taskProcessingThreadContext.isSleeping())) {
                this.getTaskItemList();
            }
        } catch (Throwable e) {
            exceptionOccurred = true;
            this.clearTaskItemAndClearTaskDataIdList();
            //如果出现了异常需要进行异常以及新的调度器分配
            if (e instanceof Exception) {
                throw (Exception) e;
            } else {
                throw new Exception(e.getMessage(), e);
            }
        }
    }
}

